/*
 * Stage.java
 */

package pipeline;

import java.util.LinkedList;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;


/**
 * A stage for the pipeline.
 *
 * Connect a data source to a data sink with a data processor piloted by one 
 * or more tasks (<code>StageTask</code>).
 *
 * @author Luca Molari
 * @param <I> the input type.
 * @param <O> the output type.
 */
public class Stage<I,O> extends Thread {
	
	protected final I_DataSource<I> source; 
	protected final I_DataSink<O> sink;
    protected final I_Processor<I,O> processor;
	
    protected final LinkedList<I_StageListener<O>> listeners;
    protected final LinkedList<StageTask<I,O>> tasks;
    
    protected ExecutorService exec;
    protected final ReentrantLock lock_get;
    protected final ReentrantLock lock_process;
    protected final ReentrantLock lock_put;
    protected final ReentrantLock lock_notify;
    
    
    /**
     * Instantiate a <code>Stage</code>.
     *
     * @param source the data source for this stage.
     * @param sink the data sink for this stage.
     * @param processor the data processor for this stage.
     * @param n_threads the number of threads for this stage.
     */
    public Stage(I_DataSource<I> source, 
    		I_DataSink<O> sink,
            I_Processor<I,O> processor,
            int n_threads) {
        
    	this.source = source; 
		this.sink = sink;
        this.processor = processor;
    	
        listeners = new LinkedList<I_StageListener<O>>();    	
    	
        tasks = new LinkedList<StageTask<I,O>>();
    	for (int i = 0; i < n_threads; ++i) {
    		StageTask<I,O> task;
    		task = new StageTask<I,O>(this);
    		tasks.add(task);
    	}
    	
    	exec = Executors.newFixedThreadPool(n_threads);
    	lock_get = new ReentrantLock(true);
    	lock_process = new ReentrantLock();
    	lock_put = new ReentrantLock();
    	lock_notify = new ReentrantLock();
    }
    
    
    /**
     * Run the stage.
     *
     */
    public void run() {
    	for (StageTask<I,O> task: tasks) {
    		exec.execute(task);
    	}
    	exec.shutdown();
    	try {
    		/* TODO: too long ? */
    		exec.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
    	}
    	catch (Exception exc) {
    	}
    	finally {
    		/* tell sink we're done */
    		sink.close();
    	}
    }

    
    /**
     * Add a listener for events in this stage.
     *
     * @param listener the new listener.
     */
    public synchronized void addStageListener(I_StageListener<O> listener) {
        if (!listeners.contains(listener)) {
            listeners.add(listener);
        }
    }

    
    /**
     * Remove a listener for events in this stage.
     *
     * @param listener the listener to remove.
     */
    public synchronized void remStageListener(I_StageListener<O> listener) {
        listeners.remove(listener);
    }

    
    /**
     * Notify all listeners for <code>onItemProcessed()</code>.
     *
     * @param listener the item just been processed.
     */
    synchronized void notifyAll_ItemProcessed(O item) {
        for (I_StageListener<O> listener : listeners) {
            listener.onItemProcessed(item);
        }
    }

    
}
